package com.youhome.analysis.config;

import com.youhome.analysis.listener.CustomerJobListener;
import com.youhome.analysis.processor.HouseItemProcessor;
import com.youhome.analysis.validator.CustomerBeanValidator;

import com.youhome.analysis.writer.HouseItemWriter;
import com.youhome.service.listhub.ListingItemType;
import com.youhome.service.listhub.ListingItemsType;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.validator.Validator;
import org.springframework.batch.item.xml.StaxEventItemReader;
import org.springframework.batch.item.xml.StaxEventItemWriter;
import org.springframework.batch.item.xml.builder.StaxEventItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.PathResource;
import org.springframework.oxm.jaxb.Jaxb2Marshaller;
import org.springframework.oxm.xstream.XStreamMarshaller;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import javax.xml.XMLConstants;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBElement;
import javax.xml.bind.Unmarshaller;
import javax.xml.transform.Source;
import javax.xml.transform.stream.StreamSource;
import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
import java.io.File;
import java.util.HashMap;
import java.util.Map;


/**
 * create by jack 2017/10/28
 * 配置类
 */
//@Configuration
//@EnableBatchProcessing
public class HouseBatchConfig {

    @Autowired
    private HouseItemWriter houseItemWriter;

    /**
     * ItemReader定义,用来读取数据
     * 1，使用FlatFileItemReader读取文件
     * 2，使用FlatFileItemReader的setResource方法设置csv文件的路径
     * 3，对此对cvs文件的数据和领域模型类做对应映射
     *
     * @return
     * @throws Exception
     */
    @Bean
    public ItemReader<ListingItemType> reader() throws Exception {

        XStreamMarshaller xStreamMarshaller = new XStreamMarshaller();
        xStreamMarshaller.setAnnotatedClasses(ListingItemType.class);
        StaxEventItemReader<ListingItemType> xmlItemReader = new StaxEventItemReaderBuilder<ListingItemType>()
                .name("xmlItemReader")
                //.resource(new ClassPathResource("/alldata/listhub/data/example.xml"))
                .resource(new PathResource("C:\\alldata\\listhub\\data\\example.xml"))
                .addFragmentRootElements("Listing")
                .unmarshaller(xStreamMarshaller)
                .build();
        return xmlItemReader;


        /*

        Jaxb2Marshaller jaxb2Marshaller = new Jaxb2Marshaller();
        jaxb2Marshaller.setClassesToBeBound(ListingItemsType.class);
        jaxb2Marshaller.setSchema(new ClassPathResource("/alldata/listhub/Syndication/2012-03/Syndication.xsd"));
        jaxb2Marshaller.setLazyInit(true);
        //jaxb2Marshaller.setPackagesToScan("Listing");
        StaxEventItemReader<ListingItemType> xmlItemReader = new StaxEventItemReaderBuilder<ListingItemType>()
                .name("xmlItemReader")
                //.resource(new ClassPathResource("/alldata/listhub/data/example.xml"))
                .resource(new PathResource("C:\\alldata\\listhub\\data\\example.xml"))

                .addFragmentRootElements("Listing")
                .unmarshaller(jaxb2Marshaller)
                .build();

        return xmlItemReader;

            */
/*
        File xsdFile = new File("/alldata/listhub/Syndication/2012-03/Syndication.xsd");
        File xmlFile = new File("/alldata/listhub/data/example.xml");
        JAXBContext context = JAXBContext.newInstance(ListingItemsType.class);
        Unmarshaller unmarshaller = context.createUnmarshaller();
        unmarshaller.setSchema(SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI).newSchema(xsdFile));
        JAXBElement<ListingItemsType> root = unmarshaller.unmarshal(new StreamSource(xmlFile), ListingItemsType.class);

        return root;

        */


    }


    /**
     * ItemProcessor定义，用来处理数据
     *
     * @return
     */
    @Bean
    public ItemProcessor<ListingItemType, ListingItemType> processor() {
        //使用我们自定义的ItemProcessor的实现CsvItemProcessor
        HouseItemProcessor processor = new HouseItemProcessor();
        //为processor指定校验器为CsvBeanValidator()
        processor.setValidator(customerBeanValidator());
        return processor;
    }

    /**
     * JobRepository，用来注册Job的容器
     * jobRepositor的定义需要dataSource和transactionManager，Spring Boot已为我们自动配置了
     * 这两个类，Spring可通过方法注入已有的Bean
     *
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws
            Exception {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDataSource(dataSource);
        jobRepositoryFactoryBean.setTransactionManager(transactionManager);
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        return jobRepositoryFactoryBean.getObject();
    }

    /**
     * JobLauncher定义，用来启动Job的接口
     *
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws
            Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
        return jobLauncher;
    }

    /**
     * Job定义，我们要实际执行的任务，包含一个或多个Step
     *
     * @param jobBuilderFactory
     * @param s1
     * @return
     */
    @Bean
    public Job importJob(JobBuilderFactory jobBuilderFactory, Step s1) {
        return jobBuilderFactory.get("importJob")
                .incrementer(new RunIdIncrementer())
                .flow(s1)//为Job指定Step
                .end()
                .listener(customerJobListener())//绑定监听器csvJobListener
                .build();
    }

    /**
     * step步骤，包含ItemReader，ItemProcessor和ItemWriter
     *
     * @param stepBuilderFactory
     * @param reader
     * @param writer
     * @param processor
     * @return
     */
    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<ListingItemType> reader,
                      ItemWriter<ListingItemType> writer, ItemProcessor<ListingItemType, ListingItemType> processor) {
        return stepBuilderFactory
                .get("step1")
                .<ListingItemType, ListingItemType>chunk(65000)//批处理每次提交65000条数据
                .reader(reader)//给step绑定reader
                .processor(processor)//给step绑定processor
                .writer(houseItemWriter)//给step绑定writer
                .build();
    }

    @Bean
    public CustomerJobListener customerJobListener() {
        return new CustomerJobListener();
    }

    @Bean
    public Validator<ListingItemType> customerBeanValidator() {
        return new CustomerBeanValidator<>();
    }


}
